package cn.tang.tframe.mq.kafka;

import cn.tang.tframe.common.annotation.QtSpi;
import cn.tang.tframe.common.utils.lang.StringUtils;
import cn.tang.tframe.mq.IQtMqFactory;
import cn.tang.tframe.mq.consumer.IQtMsgProcessor;
import cn.tang.tframe.mq.kafka.consumer.KafkaConsumerClient;
import cn.tang.tframe.mq.kafka.producer.KafkaProducerClient;
import cn.tang.tframe.mq.producer.IQtProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

/**
 * Created by tanghc on 17/1/20.
 */
@QtSpi("kafka")
public class KafkaMqFactory implements IQtMqFactory {
	static Logger logger = LoggerFactory.getLogger(KafkaMqFactory.class);


	@Override
	public IQtProducer builderPruduer() {
		return KafkaProducerClient.getQtProducer();
	}

	@Override
	public Runnable builderConsumer(String rawTopic,IQtMsgProcessor msgProcessor) {
		return builderConsumer(rawTopic,null,msgProcessor);
	}

	@Override
	public Runnable builderConsumer(String projectCode, String appCode, String topic,IQtMsgProcessor msgProcessor) {
		return builderConsumer(projectCode, appCode, topic,null,msgProcessor);
	}

	@Override
	public Runnable builderConsumer(String rawTopic, Properties properties,IQtMsgProcessor msgProcessor) {
		if(StringUtils.isBlank(rawTopic)){
			throw new RuntimeException("empty topic exception");
		}
		return KafkaConsumerClient.createConsumerTask(rawTopic,msgProcessor,properties);
	}

	@Override
	public Runnable builderConsumer(String projectCode, String appCode, String topic, Properties properties,IQtMsgProcessor msgProcessor) {
		String rawTopic = StringUtils.join(new Object[]{projectCode,appCode,topic},"_");
		return builderConsumer(rawTopic,properties,msgProcessor);
	}
}
